package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
    TODO RDD -- mapPartitions转换算子
      1、mapPartitions方法（算子），可以以分区为单位进行数据的处理，处理的逻辑会调用分区数量的次数。
      2、既然以分区为单位，所以处理的数据就是以分区的数据集合，处理后返回的也应该是数据集合，要返回一个【迭代器】
      3、mapPartitions性能会比map高一些，主要体现在资源损耗上，mapPartitions的资源消耗是以分区为单位，map消耗资源是以数据为单位；
      4、仅仅从数据处理的角度，两者区别不大，当内存资源不够的情况，mapParititons不推荐使用，因为会依托当前内存，map在处理数据时，处理完之后，数据会被释放，所以推荐使用map
      5、mapPartitions这个算子，从使用的角度来讲，传递一个迭代器，所以无需考虑数据的个数。
      6、但是map这个算子，从使用角度来讲，传递一个数据，返回一个数据，数据不能丢失，不能变多的。
 */

object Spark_RDD_Pper_Transform_MapPartitions {
  def main(args: Array[String]): Unit = {


    //TODO 1、获得连接
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 2、业务逻辑 求每个分区的最大值
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
    //2.1 使用mapPartitions
    val MaxValues: RDD[Int] = rdd.mapPartitions(
      //2.2 这里注意传进去一个迭代器，返回一个迭代器，而不是单一的数据
      iter => {
        List(iter.max).iterator
      }
    )

    MaxValues.collect().foreach(println)
    //TODO 2、关闭连接
    sc.stop()
  }
}